package com.jai.pms.message.config;

import com.jai.pms.message.handle.RedisStreamErrorHandler;
import com.jai.pms.message.listener.RedisStreamListener;
import com.ruoyi.common.redis.service.RedisService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;

import java.time.Duration;

@Configuration
public class RedisStreamConfig {
    private static Logger logger = LoggerFactory.getLogger(RedisStreamListener.class);

    @Autowired
    private RedisService redisService;

    @Autowired
    private ExecutorConfig executor;

    @Autowired
    private RedisStreamListener redisStreamListener;

    @Bean(initMethod = "start", destroyMethod = "stop")
    public StreamMessageListenerContainer streamMessageListenerContainer() {
        redisService.initStream();

        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options =
                StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                        .builder()
                        .batchSize(1)
                        .executor(this.executor.executor())
                        .errorHandler(new RedisStreamErrorHandler())
                        .pollTimeout(Duration.ZERO)
                        .serializer(new StringRedisSerializer())
                        .build();

        StreamMessageListenerContainer<String, MapRecord<String, String, String>> container =
                StreamMessageListenerContainer.create(redisService.redisTemplate.getConnectionFactory(), options);

        for (int i = 0; i < executor.getCount(); i++) {
            container.receive(Consumer.from(RedisService.STREAM_GROUP, "consumer-" + i),
                    StreamOffset.create(RedisService.STREAM_KEY, ReadOffset.lastConsumed()),
                    redisStreamListener);
        }

        return container;
    }
}

